package com.tzy.rpc.transport.netty.client;/**
 * @description:
 * @Author HP
 * @create 2022/3/14 20:47
 */

import com.tzy.rpc.entity.RpcRequest;
import com.tzy.rpc.entity.RpcResponse;
import com.tzy.rpc.factory.SingletonFactory;
import com.tzy.rpc.serializer.CommonSerializer;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * @author HP
 * @create 2022-03-14 20:47
 * @desc
 **/
public class NettyClientHandler extends SimpleChannelInboundHandler<RpcResponse> {

    private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);

    private final UnprocessedRequests unprocessedRequests;

    public NettyClientHandler(){
        this.unprocessedRequests= SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    private int beatTmp=0;

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("过程调用时有错误发生:");
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
        try {
            logger.info(String.format("客户端接收到消息: %s", msg));
            //将响应数据取出
            unprocessedRequests.complete(msg);
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        int beatCnt = 5;
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                if (beatTmp++<beatCnt){
                    logger.info("发送心跳包[{}]{} ", ctx.channel().remoteAddress(),beatTmp);
                    Channel channel = ChannelProvider.get((InetSocketAddress) ctx.channel().remoteAddress(),
                            CommonSerializer.getByCode(CommonSerializer.PROTOBUF_SERIALIZER));
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setHeartBeat(true);

                    //设置一个Listener监测服务端是否接收到心跳包，如果接收到就代表对方在线，不用关闭Channel
                    channel.writeAndFlush(rpcRequest).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                }else {
                    // 打印触发了一次超时警告
                    System.out.println(ctx.channel().remoteAddress() + "超时事件：" +state.toString());

                    // 当读超时超过 3 次，我们就端口该客户端的连接
                    // 注：读超时超过 3 次，代表起码有 4 次 3s 内客户端没有发送心跳包或普通数据包
                    System.out.println(" [client]发送心跳包超过10次，关闭连接，释放更多资源");
                    ctx.channel().close(); // 手动断开连接
                }

            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
